package com.opdar.athena.message.agent;

import com.opdar.athena.message.agent.client.Client;
import com.opdar.athena.message.agent.client.ClientPool;
import com.opdar.athena.message.agent.client.ClientTask;
import com.opdar.athena.registry.ZookeeperRegistry;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * 同步/注册链路
 * Created by shiju on 2017/7/3.
 */
public class ServerListener implements Watcher,Runnable {
    private ZookeeperRegistry registry;
    private boolean isStop = false;
    private ExecutorService executor = Executors.newCachedThreadPool();
    @Autowired
    private ClientPool clientPool;
    private static final Object lock = new Object();

    public ServerListener(ZookeeperRegistry registry) {
        this.registry = registry;
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        synchronized (lock){
            connect();
            lock.notify();
        }
    }

    public void connect() {
        registry.read();
        List<InetSocketAddress> newAddress = registry.getAddresses();
        List<ClientTask> clientTaskList = new ArrayList<ClientTask>();
        for(InetSocketAddress address:newAddress){
            try {
                Client client = clientPool.getClient(address);
                //获取配置
                //服务器配置改存data内，开启线程获取zookeeper的data并开始连接服务器，获取一个栅栏，如果连接成功则释放1个栅栏，不成功释放栅栏优先其他链接或等待N秒后继续连接。主线程通过join等待所有连接返回。
                //连接服务器
                byte[] data = registry.getData(address.getAddress().getHostAddress() + ":" + address.getPort());
                DataInputStream dataInputStream = new DataInputStream(new ByteArrayInputStream(data));
                String project = dataInputStream.readUTF();
                int version = dataInputStream.readInt();
                if(client == null){
                    clientTaskList.add(new ClientTask(address,project,version));
                }else if(!client.isOpen()){//if client != null
                    clientPool.removeClient(address);
                    clientTaskList.add(new ClientTask(address,project,version));
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        try {
            List<Future<Client>> all = executor.invokeAll(clientTaskList);
            for(Future<Client> clientFuture:all){
                try{
                    Client _client = clientFuture.get();
                    clientPool.putClient(_client.getAddress(),_client);
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        while (!isStop){
            try {
                synchronized (lock){
                    registry.existsModule(this);
                    lock.wait();
                }

            }catch (KeeperException e){
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
            catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    public void start(){
        new Thread(this).start();
    }
}
